package com.atguigu.realtime.function;

import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import com.atguigu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> {

    private ThreadPoolExecutor threadPool;


    @Override
    public void open(Configuration parameters) throws Exception {
        // 获得线程池
        threadPool = ThreadPoolUtil.getThreadPool();
    }
        //读取维度信息
  public abstract    void addDim(Jedis jedis, Connection phoenixConn, T input, ResultFuture<T> resultFuture);
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        //线程交给线程池执行
        threadPool.submit(new Runnable() {

            @Override
            public void run() {

                 Jedis jedis = null;
                 Connection phoenixConn = null;

                // redis 客户端 phoenix连接对象，这些必须一个异步请求，使用一个单独的客户端
                try {
                    // 获得数据库连接
                    jedis = RedisUtil.getRedisClient();
                    phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);

                    // 使用客户端开始具体的查询数据，补充维度
                addDim(jedis, phoenixConn,input,resultFuture);



                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    //关闭资源
                    if (jedis != null) {
                        jedis.close();
                    }
                    if (phoenixConn != null) {
                        try {
                            phoenixConn.close();
                        } catch (SQLException throwables) {
                            throwables.printStackTrace();
                        }
                    }
                }


            }
        });
        }

}
